package com.roy.KafkaTest;

import java.time.Instant;
import java.time.ZoneId;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class MyProducer {

	private static final String BOOTSTRAP_SERVERS = "172.16.48.10:9092,172.16.48.11:9092,172.16.48.12:9092";
	private static final String TOPIC = "topic1";
	public static void main(String[] args) {
		Properties props = new Properties();  
        // 此处配置的是kafka的端口  
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);  
//        props.put("zk.connect", "192.168.100.218:2181");    
  
        // 配置value的序列化类  
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 配置key的序列化类  
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 	是否需要返回，可设置 all,-1,0,1
        // 0：发出去就不管了。 不保证消息发送成功。这时retries设置将不会生效。 
        // all=-1:最强有力的保证。leader发送消息后会等待至少有一个同步副本保持活动状态，记录不会丢失。
        // 1: leader将记录写入本地日志，但不会等待所有followers完全确认。在这种情况下，如果leader在确认记录后立即失败，但在followers复制记录之前，记录将丢失。
//        props.put("request.required.acks", "-1"); 
        props.put("request.required.acks", "0"); 
        
//        Producer<String,String> producer = new KafkaProducer<>(props);
//        for(int i = 0; i < 5; i++) {
//        	producer.send(new ProducerRecord<String,String>(TOPIC, Integer.toString(i), "MyProducer:"+i));
//        	System.out.println("message "+i+" sended");
//        }

//        producer.close();
        
        new Thread(() -> {
        	Producer<String,String> producer = new KafkaProducer<>(props);
        	try {
                while (true) {
                    TimeUnit.SECONDS.sleep(1L);
                    Instant now = Instant.now();
                    String key = "service_1";
                    String value = key + "@" + toLocalTimeStr(now);
                    producer.send(new ProducerRecord<>(TOPIC, key, value));
                    System.out.println("message: "+value +" sended");
                }
            } catch (Exception e) {
                e.printStackTrace();
                producer.close();
            }
        }).start();
	}
	
	private static String toLocalTimeStr(Instant i) {
	    return i.atZone(ZoneId.systemDefault()).toLocalDateTime().toString();
	}
}
